package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Demo17Cache {
  def main(args: Array[String]): Unit = {
    /**
     * 缓存
     *
     */

    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo16Action")

    val sc = new SparkContext(conf)

    //设置checkpoint保存路径
    sc.setCheckpointDir("data/checkpoint")

    //读取学生表数据
    val linesRDD: RDD[String] = sc.textFile("data/students.txt")


    //整理取出字段
    val mapRDD: RDD[Array[String]] = linesRDD.map(_.split(","))


    val studnetRDD: RDD[(String, String, Int, String, String)] = mapRDD.map {
      case Array(id: String, name: String, age: String, gender: String, clazz: String) =>
        println("=======map============")
        (id, name, age.toInt, gender, clazz)
    }

    /**
     * 对多次使用的RDD进行缓存
     *
     */

    //缓存再内存中
    //studnetRDD.cache()
    //内存放不下放磁盘，对数据做压缩
    // studnetRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)
    //studnetRDD.persist(StorageLevel.MEMORY_AND_DISK)

    /**
     * checkpoint：将rdd的数据缓存到hdfs中，任务失败了数据也不会丢失
     * cache是将数据缓存再spark执行的服务器的内存或者磁盘上，如果任务执行失败数据就没了
     * checkpoint： 主要是再spark streaming中使用，用来保证任务的高可用
     */
    studnetRDD.checkpoint()

    //1、统计班级人数
    studnetRDD
      .map {
        case (_, _, _, _, clazz: String) =>
          (clazz, 1)
      }
      .reduceByKey(_ + _)
      .saveAsTextFile("data/clazz_num")

    println("=" * 100)

    //统计性别的人数
    studnetRDD
      .map {
        case (_, _, _, gender: String, _) =>
          (gender, 1)
      }
      .reduceByKey(_ + _)
      .saveAsTextFile("data/gender_num")


    //统计年龄的人数
    studnetRDD
      .map {
        case (_, _, age: Int, _, _) =>
          (age, 1)
      }
      .reduceByKey(_ + _)
      .saveAsTextFile("data/age_num")


    while (true) {

    }

  }

}
